#include "ssiot/msgbus.h"
#include <stdio.h>
#include <time.h>
#include <stdlib.h>
#include <string.h>
#include <chrono>
#include <thread>

#ifdef WINDOWS
#pragma comment(lib, "paho-mqtt3c.lib")
#endif


volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void* context, MQTTClient_deliveryToken dt)
{
    fprintf(stdout, "Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void* context, char* topicName, int topicLen, MQTTClient_message* message)
{
    Msgbus* msgbus = (Msgbus*)context;
    msgbus->OnMessage(topicName, message);
    MQTTClient_free(topicName);
    MQTTClient_freeMessage(&message);
    return 1;
}

void connlost(void* context, char* cause)
{
    Msgbus* msgbus = (Msgbus*)context;
    msgbus->OnConnLost(cause);
}

int IoTMsg::FromJson(const char* json)
{
    nlohmann::json obj = nlohmann::json::parse(json);

    to = obj["to"];
    from = obj["from"];
    sid = obj["sid"];
    code = obj["code"];
    msg = obj["msg"];
    data = obj["data"];

    return 0;
}

std::string IoTMsg::ToJson(void)
{
    nlohmann::json json;

    if (sid > 0) {
        json["sid"] = sid;
    }

    if (to.length() > 0) {
        json["to"] = to;
    }

    if (from.length() > 0) {
        json["from"] = from;
    }

    if (code > 0) {
        json["code"] = code;
    }

    if (msg.length() > 0) {
        json["msg"] = msg;
    }

    json["data"] = data;

    return json.dump();
}

Msgbus::Msgbus(std::string sn)
{
    sn_ = sn;
    sync_ = false;
    done_ = false;

    sid_ = time(NULL);
}

Msgbus::~Msgbus()
{

}

int Msgbus::Connect(std::string address, std::string username, std::string password )
{
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    if (MQTTClient_create(&client_, address.c_str(), sn_.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL) != MQTTCLIENT_SUCCESS){
        fprintf(stderr, "failed to create client.\n");
        return -1;
    }

    if (MQTTClient_setCallbacks(client_, this, connlost, msgarrvd, delivered) != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "failed to set callbacks \n");
        return -1;
    }

    conn_opts.keepAliveInterval = 300;
    conn_opts.cleansession = 1;

    if (username.length() > 0) {
        conn_opts.username = username.c_str();
        conn_opts.password = password.c_str();
    }

    if (MQTTClient_connect(client_, &conn_opts) != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "failed to connect\n");
        return -1;
    }

    char topc[256];
    sprintf(topc, "/iot/%s/#", sn_.c_str());

    if (MQTTClient_subscribe(client_, topc, 0) != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "failed to subscribe\n");
        return -1;
    }
    fprintf(stdout, "mqtt subscribe %s\n", topc);
    return 0;
}

void Msgbus::Disconn(void)
{
    MQTTClient_disconnect(client_, 10000);
    MQTTClient_destroy(&client_);
}

void Msgbus::OnConnLost(char* cause)
{
    fprintf(stderr, "mqtt connection lost. %s\n", cause);
}

void Msgbus::Publish(std::string topic, const char* data, int len)
{   
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    pubmsg.payload = static_cast<void*>(const_cast<char*>(data));
    pubmsg.payloadlen = len;
    pubmsg.qos = 0;
    pubmsg.retained = 0;
    if (MQTTClient_publishMessage(client_, topic.c_str(), &pubmsg, &token) != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "failed to publish message\n");
        return;
    }
    MQTTClient_waitForCompletion(client_, token, 10000L);

    fprintf(stderr, "pub: %s\n", topic.c_str());
}

void Msgbus::OnMessage(std::string topic, MQTTClient_message* message) 
{
    fprintf(stderr, "sub: %s\n", topic.c_str());

    IoTMsg iot;

    char* date = static_cast<char*>(message->payload);
    date[message->payloadlen] = 0;

    if (iot.FromJson(date) != 0){
        fprintf(stderr, "decode iot message error.\n");
        return;
    }

    if (sync_ && req_.sid == iot.sid) {
        ack_ = iot;
        done_ = true;
        return;
    }
}

void Msgbus::Publish(IoTMsg msg)
{
    if (msg.to.length() == 0) {
        msg.to = "$";
    }

    if (msg.from.length() == 0) {
        msg.from = sn_;
    }

    char topic[128];
    sprintf(topic, "/iot/%s/%s/%s", msg.to.c_str(), msg.from.c_str(), msg.type.c_str());

    if (msg.from != sn_) {
        msg.from = sn_;
    }

    std::string data = msg.ToJson();

    this->Publish(topic, data.c_str(), data.length());
}

IoTMsg Msgbus::SyncPublish(IoTMsg msg)
{
    if (sync_) {
        fprintf(stderr, "error syncing......\n");
        msg.code = -1;
        return msg;
    }

    msg.sid = sid_++;

    req_ = msg;
    done_ = false;
    sync_ = true;
    Publish(msg);

    for (int i = 0; i < 1000; i++) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        if (done_) {
            done_ = false;
            sync_ = false;
            return ack_;
        }
    }

    msg.code = -2;
    return msg;
}

